//client
#include <stdio.h>                                                             
#include <stdlib.h>                                                            
#include <pthread.h>                                                           
#include <string.h>                                                            
#include <semaphore.h>
#include <assert.h>
#include <time.h>
#include <stddef.h>
#include <sys/types.h>                                                         
#include <sys/ipc.h>                                                           
#include <sys/shm.h>      
#include <sys/msg.h>      
#include "client.h"                                                      

void * buffer;
int * producer;
int * consumer;
int msgid, shmid;
//struct timeval start,end,diff;
struct request *bufferRing;
pthread_mutex_t bufferRingMutex = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t *buffMutex;
sem_t sema;

void *clientThread(void *param);

int main(int argc, char * argv[]){                                             
  //number of threads                                                          
  //ms time                                                                    
  //cid
  //# of rounds
  if (argc != 6){
    fprintf(stderr, "usage: ./client [# of threads] [ms time] \n[client id] [# of rounds]\n[buffer size]");
    exit(-1);
  }
  int numThreads = atoi(argv[1]);
  int msPerRound = atoi(argv[2]);
  int cid = atoi(argv[3]);
  int numRounds = atoi(argv[4]);
  int buffSize = atoi(argv[5]);
  int pid = getpid();
  int counter, rc;
  key_t key, shmkey;
  struct my_msgbuf msgToSend;
  pthread_attr_t attr;
  void * status;

  srand(time(NULL));
  buffMutex = malloc(buffSize * sizeof(pthread_mutex_t));
  for (counter= 0; counter < buffSize; counter++){
    pthread_mutex_init(&(buffMutex[counter]), NULL);
  }
  /* Initialize and set thread detached attribute */
  pthread_attr_init(&attr);
  pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE);

  fprintf( stderr, "client cid: %d pid: %d\nbuffsize: %d, %d threads\n%d ms time\n%d numRounds\n", cid, pid,buffSize, numThreads, msPerRound, numRounds);
  
  //create pthread param array
  struct pthread_param *paramArray = malloc(sizeof (struct pthread_param) * numThreads);
  struct pthread_param *currentParam = paramArray;
  

  //initialize semaphore
  sem_init(&sema, 0, buffSize - 1);
  key = ftok(".",'b');
  msgid = msgget(key, 0666| IPC_CREAT);
  fprintf(stderr, "client %d got msgid %d\n", cid, msgid);
  //create shared memory
  shmkey = ftok(".", (char)cid);
  shmid = shmget(shmkey, sizeof(struct request) * buffSize + 2 * sizeof(int), 0777 | IPC_CREAT );
  fprintf(stderr, "client %d got shmid %d\n", cid, shmid);
  if (shmid == -1){
    perror("client shmget");
    exit(1);
  }
  buffer = shmat(shmid, (void *) NULL, 0);
  if (buffer == (void *)-1){
    perror("client shmat");
    exit(1);
  }

  //set producer, consumer, bufferRing pointers
  producer = buffer;
  consumer = (int *)buffer + 1;
 
  //send a message with the information
  msgToSend.mtype = 1;
  msgToSend.shmid = shmid;
  if (msgsnd(msgid, &msgToSend,
	     sizeof(int), 0) == -1){
    perror("client msgsnd");
    exit(1);
  }
  //move pointer to point to first request array in buffer
  bufferRing = (struct request *)((int*)buffer + 2);
  *producer= -1;//setting producer for server 
  //initialize the bufferRing
  for (counter =0; counter < buffSize; counter++){
    bufferRing[counter].result = -1;
    bufferRing[counter].buffSize = buffSize;
  }

  //launch the threads
  for (counter =0; counter < numThreads; counter++){
    //prepare pthread_param structure
    currentParam->numThreads = numThreads;
    currentParam->msPerRound = msPerRound;
    currentParam->cid = cid;
    currentParam->numRounds = numRounds;
    currentParam->pid = pid;
    currentParam->buffSize = buffSize;
    rc = pthread_create(&(currentParam->tid), 
			&attr, 
			clientThread, 
			(void *)(currentParam));
    if (rc){
      fprintf(stderr, "error from pthread create %d\n", rc);
    }
    currentParam++;
  }

  //join the threads
  currentParam = paramArray;
  for (counter =0; counter < numThreads; counter++){
    rc = pthread_join(currentParam->tid, &status);
    if (rc){
      perror("client pthread_join");
      exit(1);
    }
    else{
    fprintf(stderr, "joined pthread %d", counter);
    }
    currentParam++;
  }
  *producer = -2;

}


buff_wait(struct pthread_param *paramPtr){
  //add new request
  int currentSpot = 0;
  int servicetime=0;
  struct timeval start,end,diff;
  int buffSize = paramPtr->buffSize;
  //sem_wait(&sema);
 lockup:    pthread_mutex_lock(&bufferRingMutex);
  //add a new value in request
    if (*producer == -1){ //producer uninitialized first request placed
      pthread_mutex_lock(&buffMutex[0]);	
      bufferRing[0].input =(rand()%10+1);
      bufferRing[0].p =(rand()%10+1);
      fprintf(stderr,"Client %d thread %d placed request x=%d p=%d\n at spot 0",paramPtr->cid,paramPtr->tid,bufferRing[0].input,bufferRing[0].p);
      currentSpot =0;
      *producer= 1; //modified producer to 0 instead of 1
    }
    else if ((((*producer) + 1) % buffSize) ==*consumer){//cannot go further
      pthread_mutex_unlock(&bufferRingMutex);
      usleep(1000);
      goto lockup; //wait for client to place request
    }
    else{ //can go further
      if (!pthread_mutex_trylock(&buffMutex[*producer])){//it locked
  	bufferRing[*producer].input =(rand()%10+1);
	bufferRing[*producer].p =(rand()%10+1);
	fprintf(stderr,"Client %d thread %d placed request x=%d p=%d at spot %d\n",paramPtr->cid,paramPtr->tid,bufferRing[*producer].input,bufferRing[*producer].p,*producer);
	currentSpot = *producer;
	*producer = (*producer + 1)%buffSize;
      }
      else{
	pthread_mutex_unlock(&bufferRingMutex);
	usleep(10000);
	goto lockup;
      }		
    }
    //continue checking whether the result has arrived and consumer has passed you
    pthread_mutex_unlock(&bufferRingMutex);
   gettimeofday(&start,NULL); 
   while (bufferRing[currentSpot].result == -1){
     //usleep(1000);
	//fprintf(stderr,"point 1 inside buffwait at spot %d\n",currentSpot);
    }
    assert(bufferRing[currentSpot].result != -1);
    fprintf(stderr,"Thread %d in client %d received result: %ld\n",paramPtr->tid, paramPtr->cid, bufferRing[currentSpot].result);
    gettimeofday(&end,NULL);
    timeval_subtract(&diff,&end,&start);
    servicetime= diff.tv_sec * 1000000 + diff.tv_usec ;
    fprintf(stderr,"Time to service thread %d : %d us \n",paramPtr->tid,servicetime);
    bufferRing[currentSpot].result = -1;  //set result back to -1
    assert(bufferRing[currentSpot].result == -1);
    fprintf(stderr, "\nthread id %d out at %d\n", paramPtr->tid, currentSpot);
    pthread_mutex_unlock(&(buffMutex[currentSpot]));
    //sem_post(&sema);


}
void * clientThread(void * param){
  fprintf(stderr, "in pthread\n");
  struct pthread_param * paramPtr = param;
  struct timeval results,now,lastUpdate; 
  int msElapsed = 0;
  int counter =0;
  int buffSize = paramPtr->buffSize;
  int currentSpot = 0;
  int valp;

  gettimeofday(&lastUpdate, NULL);
  while (counter < paramPtr->numRounds){
    usleep(1000);
    gettimeofday(&now, NULL);
    timeval_subtract(&results, &now, &lastUpdate);
    msElapsed += results.tv_sec * 1000 + results.tv_usec / 1000;
    lastUpdate = now;
    if (msElapsed < paramPtr->msPerRound){
      continue;
    }
    buff_wait(paramPtr);
    //done with request, prepare for next round
    msElapsed = 0;
    gettimeofday(&lastUpdate, NULL);
    counter++;
    fprintf(stderr, "finished a round\n");
  }
  fprintf(stderr, "exiting pthread\n");
  pthread_exit(NULL);
  
  
}


timeval_subtract (struct timeval *result, struct timeval *x, struct timeval *y)
{
  /* Perform the carry for the later subtraction by updating y. */
  if (x->tv_usec < y->tv_usec) {
    int nsec = (y->tv_usec - x->tv_usec) / 1000000 + 1;
    y->tv_usec -= 1000000 * nsec;
    y->tv_sec += nsec;
  }
  if (x->tv_usec - y->tv_usec > 1000000) {
    int nsec = (x->tv_usec - y->tv_usec) / 1000000;
    y->tv_usec += 1000000 * nsec;
    y->tv_sec -= nsec;
  }

  /* Compute the time remaining to wait.                                                                                                                                        
     tv_usec is certainly positive. */
  result->tv_sec = x->tv_sec - y->tv_sec;
  result->tv_usec = x->tv_usec - y->tv_usec;

  /* Return 1 if result is negative. */
  return x->tv_sec < y->tv_sec;
}


        
